package org.leafDomain.rdb.impl;

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.leafDomain.rdb.RDB;

import java.sql.*;
import java.util.*;

public class RDBImpl implements RDB {

    private static final Logger loggerRDB = Logger.getLogger(RDBImpl.class);
    private final Properties config;
    private Connection con;
    private PreparedStatement pst;
    private ResultSet rst;
    private int batchSize;
    private String hbaseRowKeyRdbCol;
    Map<String, Map<String, String>> hbaseRdbColumnsMapping;

    /**
     * 外部的资源向内传
     * @param config 配置文件信息
     */
    public RDBImpl(Properties config) {
        this.config = config;
    }

    @Override
    public void init() throws Exception {
        con = getConnection();
        loggerRDB.info("RDB 创建[ 连接 ]对象成功");
        pst = getStatement(con);
        loggerRDB.info("RDB 创建[ 执行 ]对象成功");
        rst = getResult(pst);
        loggerRDB.info("RDB 创建[ 结果集 ]成功");
        batchSize = batchSize();
        hbaseRdbColumnsMapping = hbaseRdbColumnsMapping();
    }

    @Override
    public void close() {
        closeAll(rst, pst, con);
    }

    @Override
    public boolean hashNextBatch() throws SQLException{
        return rst.next();
    }

    @Override
    public List<Put> nextBatch() throws SQLException{
        List<Put> list = new ArrayList<>(batchSize);
        int count = 0;
        do{
            /*
             * 将 mysql 表中的一行记录转换成多个 Put
             */
            // mysql主键作为HBase行键
            Put put = new Put(toBytesFromRdb(hbaseRowKeyRdbCol));
            for (Map.Entry<String, Map<String, String>> e : hbaseRdbColumnsMapping.entrySet()) {
                String columnFamily = e.getKey();
                for (Map.Entry<String, String> s : e.getValue().entrySet()) {
                    String hbaseColumn = s.getKey();
                    String rdbColumn = s.getValue();
                    // 列族，列名，数值
                    put.addColumn(strToBytes(columnFamily), strToBytes(hbaseColumn), toBytesFromRdb(rdbColumn));
                }
            }
            list.add(put);
        }while(++count < batchSize && rst.next());
        return list;
    }


    /**
     * 重载Com的config，用于将此类的config给Com统一检查
     * @return properties文件对象
     */
    @Override
    public Properties config() {
        return config;
    }



    /**
     * 检查并获取配置文件信息
     * @return 配置信息
     */
    private String driver() {
        return checkAndGetConfig("rdb.driver");
    }
    private String url(){
        return checkAndGetConfig("rdb.url");
    }
    private String username(){
        return checkAndGetConfig("rdb.username");
    }
    private String password(){
        return checkAndGetConfig("rdb.password");
    }
    private String sql(){
        return checkAndGetConfig("rdb.sql");
    }
    private int batchSize(){
        return Integer.parseInt(checkAndGetConfig("rdb.batchSize"));
    }


    /**
     * hbase列族和列与rdb中列的映射关系
     *             hbase列族     列     rdb表的列
     * @return Map<String, Map<String, String>>
     */
    private Map<String, Map<String, String>> hbaseRdbColumnsMapping(){
        Map<String, Map<String, String>> map = hBaseColumnsMapping("rdb.hbase.columns.mapping");
        hbaseRowKeyRdbCol = map.remove("ROWKEY").get("ROWKEY");
        return map;
    }

    /**
     * 加载数据库驱动，并建立与数据库的连接。
     *
     * @return Connection 与数据库的连接对象。
     * @throws ClassNotFoundException 如果找不到指定的数据库驱动类。
     * @throws SQLException 如果建立数据库连接时出现错误。
     */
    private Connection getConnection() throws ClassNotFoundException, SQLException {
        // 加载数据库驱动
        Class.forName(driver());
        // 通过DriverManager获取数据库连接对象并返回
        return DriverManager.getConnection(url(), username(), password());
    }


    /**
     * 根据提供的数据库连接对象创建并返回一个PreparedStatement对象，该对象用于执行预编译的SQL语句。
     *
     * @param con 数据库连接对象，用于创建PreparedStatement。
     * @return PreparedStatement 准备好的预编译SQL语句对象，可用于执行查询或更新操作。
     * @throws SQLException 如果创建PreparedStatement对象时出现SQL异常。
     */
    private PreparedStatement getStatement(Connection con) throws SQLException {
        return con.prepareStatement(sql());
    }

    /**
     * 从给定的PreparedStatement对象中执行查询并返回查询结果集。
     *
     * @param statement 预编译的SQL语句对象，包含要执行的查询。
     * @return ResultSet 包含查询结果的集合。
     * @throws SQLException 如果执行查询时出现SQL异常。
     */
    private ResultSet getResult(PreparedStatement statement) throws SQLException {
        return statement.executeQuery();
    }

    private byte[] toBytesFromRdb(String rdbColumn) throws SQLException {
        return toBytes(rst.getObject(rdbColumn));
    }

    private byte[] strToBytes(String name){
        return Bytes.toBytes(name);
    }




}
